/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.lastfm;

import aiaudio.lastfm.hbase.SchemeCommunicator;
import de.umass.lastfm.Artist;
import de.umass.lastfm.CallException;
import de.umass.lastfm.Period;
import de.umass.lastfm.User;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import aiaudio.database.tables.UsersListTable;

/**
 *
 * @author nastya
 */
public class TracksByUserExtracter {
    private SchemeCommunicator communicator;
    private int tracksPerUser;

    public TracksByUserExtracter(SchemeCommunicator communicator, 
            int tracksPerUser) {
        this.communicator = communicator;
        this.tracksPerUser = tracksPerUser;
    }

    public TracksByUserExtracter(SchemeCommunicator communicator) {
        this(communicator, 0);
    }

    
    
    public void start() throws IOException{
        HTable usersTable = communicator.getUsersTable();
        Scan scan = new Scan();
        scan.addColumn(UsersListTable.DATA.getNameAsBytes(), UsersListTable.USERNAME.getNameAsBytes());
        ResultScanner scanner = usersTable.getScanner(scan);
        Result next;
        while((next = scanner.next()) != null){
            List<KeyValue> column = next.getColumn(UsersListTable.DATA.getNameAsBytes(), UsersListTable.USERNAME.getNameAsBytes());
            String userName = Bytes.toString(column.get(0).getValue());
            try{
            Collection<Artist> topArtists = User.getTopArtists(userName, Period.OVERALL, LastFmDownloaderHBase.API_KEY);
            int cnt = 0;
            for (Artist artist : topArtists) {
                if(tracksPerUser > 0 && cnt >= tracksPerUser){
                    break;
                }
                String artistId = addArtist(artist);
                
                String rowId = Bytes.toString(next.getRow());
                addRating(rowId, artistId, artist.getPlaycount());
                artist.getPlaycount();
                cnt++;
            }
            }catch(CallException ex){
                System.out.println("        Call exception at " + userName);
            }
        }
    }

    private String addArtist(Artist artist) throws IOException {
        return communicator.insertArtist(artist);
    }

    private void addRating(String userId, String artistId, int playcount) throws IOException {
        communicator.insertRating(userId, artistId, playcount);
    }
}
